/*
 * @(#)LocalFileCanalConfig.java   1.0  2021年11月4日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.source.canal;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.otter.canal.instance.manager.CanalConfigClient;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.IndexMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.RunMode;
import com.alibaba.otter.canal.instance.manager.model.CanalStatus;
import com.swift.core.env.ConfigInterface;
import com.swift.core.env.EnvDecode;
import com.swift.dao.db.datasource.DataBase;
import com.swift.dts.dao.TablesDao;
import com.swift.dts.loadconfig.model.DtsDataBase;
import com.swift.exception.extend.SystemException;
import com.swift.util.exec.PathUtil;
import com.swift.util.type.TypeUtil;

/**
 * 添加说明 
 * @author zhengjiajin
 * @version 1.0 2021年11月4日
 */
@Component
public class LocalFileCanalConfig extends CanalConfigClient {
    
    public static final String CANAL_PATH = "data";
    
    @Autowired
    private ConfigInterface<DtsDataBase> dataBaseConfigInterface;
    //@Autowired
    //private ConfigInterface<TablesConfig> tablesConfigInterface;
    
    private AtomicLong canalId = new AtomicLong(0);
    
    private AtomicLong slaveId = new AtomicLong(0);
    
    private Map<String,Canal> canalMap = new ConcurrentHashMap<>();
    
    /**
     * 根据对应的destinantion查询Canal信息
     */
    public Canal findCanal(String destination) {
        Canal canal = null;
        if(canalMap.containsKey(destination)) canal = canalMap.get(destination);
        else canal = new Canal();
        if(canal.getId()==null) canal.setId(canalId.incrementAndGet());
        
        CanalParameter canalParameter = null;
        if(canal.getCanalParameter()==null) {
            canalParameter = new CanalParameter();
            canal.setCanalParameter(canalParameter);
        }
        else  canalParameter = canal.getCanalParameter();
        if(canal.getGmtCreate()==null) canal.setGmtCreate(new Date());
        canal.setGmtModified(new Date());
        canal.setName(destination);
        canal.setStatus(CanalStatus.START);
        canalParameter.setCanalId(canal.getId());
        canalParameter.setRunMode(RunMode.EMBEDDED);
        canalParameter.setDataDir(PathUtil.findOnClassPath(CANAL_PATH));// 默认本地文件数据的目录默认是data
        canalParameter.setMetaMode(MetaMode.LOCAL_FILE);
        canalParameter.setDetectingSQL("select 1");// 现只支持mysql
        canalParameter.setDetectingIntervalInSeconds(3);
        canalParameter.setDetectingTimeoutThresholdInSeconds(30); // 心跳超时时间
        canalParameter.setDetectingRetryTimes(3); // 心跳检查重试次数
        canalParameter.setIndexMode(IndexMode.META);
        canalParameter.setMemoryStorageRawEntry(false);
        canalParameter.setDbAddresses(getDataBaseUrl(destination));
        //canalParameter.setMasterAddress(getDataBaseUrl(destination).get(0));
        canalParameter.setDbUsername(getDataBaseUsername(destination));
        canalParameter.setDbPassword(getDataBasePassword(destination));
        if(canalParameter.getSlaveId()==null)canalParameter.setSlaveId(slaveId.incrementAndGet());
        canalParameter.setDefaultDatabaseName(getDefaultDatabaseName(destination));
        //canalParameter.setBlackFilter(findBlackFilter(destination));
        canalParameter.setBlackFilter(TablesDao.SYS_DB_TABLE);
        canalParameter.setDdlIsolation(Boolean.TRUE);
        canalMap.put(destination, canal);
        return canal;
    }
    //支持多库
    private List<InetSocketAddress> getDataBaseUrl(String destination) {
        DataBase database = dataBaseConfigInterface.findConfig(destination);
        if(TypeUtil.isNull(database.getUrl())) throw new SystemException("配置的"+destination+"连接异常");
        List<InetSocketAddress> addressList = new ArrayList<>();
        for(String url:database.getUrl()) {
            try {
                InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(getAddress(url)), getPort(url));
                addressList.add(address);
            } catch (UnknownHostException e) {
                throw new SystemException("配置的"+destination+"连接"+url+"异常");
            }
        }
        return addressList;
    }
    //只支持相同的帐号密码
    private String getDataBaseUsername(String destination) {
        DataBase database = dataBaseConfigInterface.findConfig(destination);
        if(TypeUtil.isNotNull(database.getUserName())) return EnvDecode.decode(database.getUserName()[0]);
        throw new SystemException("配置的"+destination+"帐号异常");
    }
    
    //只支持相同的帐号密码
    private String getDataBasePassword(String destination) {
        DataBase database = dataBaseConfigInterface.findConfig(destination);
        if(TypeUtil.isNotNull(database.getPassword())) return EnvDecode.decode(database.getPassword()[0]);
        throw new SystemException("配置的"+destination+"密码异常");
    }
    
    //只支持相同的帐号密码
    private String getDefaultDatabaseName(String destination) {
        DataBase database = dataBaseConfigInterface.findConfig(destination);
        if(TypeUtil.isNull(database.getUrl())) throw new SystemException("配置的"+destination+"连接异常");
        String url = database.getUrl()[0];
        //jdbc:mysql://rm-wz92fx5fq01u61gj4oo.mysql.rds.aliyuncs.com:3306/hhmk_dev?useUnicode=true&characterEncoding=utf8
        return getDb(url);
       
    }
    
    private String cleanUrl(String url) {
        if(url.indexOf("//")!=-1)
            url=url.substring(url.indexOf("//")+2);
        if(url.indexOf("?")!=-1)
            url=url.substring(0,url.indexOf("?"));
        return url;
    }
    
    private String getAddress(String url) {
        url=cleanUrl(url);
        if(url.indexOf(":")!=-1)
            url=url.substring(0,url.indexOf(":"));
        if(url.indexOf("/")!=-1)
            url=url.substring(0,url.indexOf("/"));
        return url;
    }
    
    private int getPort(String url) {
        url=cleanUrl(url);
        if(url.indexOf(":")!=-1)
            url=url.substring(url.indexOf(":")+1);
        if(url.indexOf("/")!=-1)
            url=url.substring(0,url.indexOf("/"));
        if(url.indexOf("?")!=-1)
            url=url.substring(0,url.indexOf("?"));
        return TypeUtil.toInt(url, 3306);
    }
    
    private String getDb(String url) {
        url=cleanUrl(url);
        if(url.indexOf("/")!=-1)
            url=url.substring(url.indexOf("/")+1);
        if(url.indexOf("?")!=-1)
            url=url.substring(0,url.indexOf("?"));
        return url;
    }
    
    /**
     * 根据对应的destinantion查询filter信息
     */
    /*@Override
    public String findFilter(String destination) {
        TablesConfig config = tablesConfigInterface.findConfig(TablesDao.synTable);
        if(config==null) return "";
        return FormatRegexUtil.formatRegex((List<String>)config.getRegex());
    }*/
    
    @Override
    public String findFilter(String destination) {
        return ".*";
    }
    
    /**
     * 根据对应的destinantion查询filter信息
     */
    /*@SuppressWarnings("unchecked")
    public String findBlackFilter(String destination) {
        TablesConfig config = tablesConfigInterface.findConfig(TablesDao.blackFilter);
        if(config==null) return "";
        return FormatRegexUtil.formatRegex((List<String>)config.getRegex());
    }*/

}
